package com.test.mq;  
  
import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;  
  
public class Consumer2 {  
    public static void main(String[] args){  
        DefaultMQPushConsumer consumer =   
                new DefaultMQPushConsumer("PullConsumer");  
        consumer.setNamesrvAddr("192.168.0.105:9876;192.168.0.106:9876");   
        try {  
            //订阅PushTopic下Tag为push的消息  
            consumer.subscribe("PullTopic", "pull");  
            //程序第一次启动从消息队列头取数据  
            consumer.setConsumeFromWhere(  
                    ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);  
            consumer.registerMessageListener(  
                new MessageListenerConcurrently() {  
                    public ConsumeConcurrentlyStatus consumeMessage(  
                            List<MessageExt> list,  
                            ConsumeConcurrentlyContext Context) {  
                        Message msg = list.get(0);  
                        System.out.println(msg.toString());  
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;  
                    }  
                }  
            );  
            consumer.start();  
        } catch (Exception e) {  
            e.printStackTrace();  
        }  
    }  
}  